﻿/*******************************************************
 *
 * CLR版本：4.0.30319.42000
 * 说明：
 *
 * 历史记录：
 *     创建文件 Luomg 2021-08-23
 *
*******************************************************/
using System;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Azure.ServiceBus;
using Microsoft.Extensions.Logging;

namespace EventBus.Azure {
    public class EventBusServiceBus:IEventBus {
        private readonly IServiceBusPersisterConnection _serviceBusPersisterConnection;
        private readonly ILogger<EventBusServiceBus> _logger;
        private readonly IEventBusSubscriptionsManager _subsManager;
        private readonly IServiceScopeFactory _serviceScopeFactory;
        private readonly string AUTOFAC_SCOPE_NAME = "eshop_event_bus";
        private const string INTEGRATION_EVENT_SUFFIX = "IntegrationEvent";
        public EventBusServiceBus(IServiceBusPersisterConnection serviceBusPersisterConnection,
            ILogger<EventBusServiceBus> logger,IEventBusSubscriptionsManager subsManager,IServiceScopeFactory serviceScopeFactory) {
            _serviceBusPersisterConnection = serviceBusPersisterConnection;
            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
            _subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager();
            _serviceScopeFactory = serviceScopeFactory;
            RemoveDefaultRule();
            RegisterSubscriptionClientMessageHandler();
        }
        public void Publish(IntegrationEvent @event) {
            var eventName = @event.GetType().Name.Replace(INTEGRATION_EVENT_SUFFIX,"");
            var jsonMessage = JsonSerializer.Serialize(@event);
            var body = Encoding.UTF8.GetBytes(jsonMessage);
            var message = new Message {
                MessageId = Guid.NewGuid().ToString(),
                Body = body,
                Label = eventName,
            };
            _serviceBusPersisterConnection.TopicClient.SendAsync(message)
                .GetAwaiter()
                .GetResult();
        }
        public void SubscribeDynamic<TH>(string eventName)
            where TH : IDynamicIntegrationEventHandler {
            _logger.LogInformation("Subscribing to dynamic event {EventName} with {EventHandler}",eventName,typeof(TH).Name);
            _subsManager.AddDynamicSubscription<TH>(eventName);
        }
        public void Subscribe<T, TH>()
            where T : IntegrationEvent
            where TH : IIntegrationEventHandler<T> {
            var eventName = typeof(T).Name.Replace(INTEGRATION_EVENT_SUFFIX,"");
            var containsKey = _subsManager.HasSubscriptionsForEvent<T>();
            if(!containsKey) {
                try {
                    _serviceBusPersisterConnection.SubscriptionClient.AddRuleAsync(new RuleDescription {
                        Filter = new CorrelationFilter { Label = eventName },
                        Name = eventName
                    }).GetAwaiter().GetResult();
                } catch(ServiceBusException) {
                    _logger.LogWarning("The messaging entity {eventName} already exists.",eventName);
                }
            }
            _logger.LogInformation("Subscribing to event {EventName} with {EventHandler}",eventName,typeof(TH).Name);
            _subsManager.AddSubscription<T,TH>();
        }
        public void Unsubscribe<T, TH>()
            where T : IntegrationEvent
            where TH : IIntegrationEventHandler<T> {
            var eventName = typeof(T).Name.Replace(INTEGRATION_EVENT_SUFFIX,"");
            try {
                _serviceBusPersisterConnection
                    .SubscriptionClient
                    .RemoveRuleAsync(eventName)
                    .GetAwaiter()
                    .GetResult();
            } catch(MessagingEntityNotFoundException) {
                _logger.LogWarning("The messaging entity {eventName} Could not be found.",eventName);
            }
            _logger.LogInformation("Unsubscribing from event {EventName}",eventName);
            _subsManager.RemoveSubscription<T,TH>();
        }
        public void UnsubscribeDynamic<TH>(string eventName)
            where TH : IDynamicIntegrationEventHandler {
            _logger.LogInformation("Unsubscribing from dynamic event {EventName}",eventName);
            _subsManager.RemoveDynamicSubscription<TH>(eventName);
        }
        public void Dispose() {
            _subsManager.Clear();
        }
        private void RegisterSubscriptionClientMessageHandler() {
            _serviceBusPersisterConnection.SubscriptionClient.RegisterMessageHandler(
                async (message,token) => {
                    var eventName = $"{message.Label}{INTEGRATION_EVENT_SUFFIX}";
                    var messageData = Encoding.UTF8.GetString(message.Body);

                    if(await ProcessEvent(eventName,messageData)) {
                        await _serviceBusPersisterConnection.SubscriptionClient.CompleteAsync(message.SystemProperties.LockToken);
                    }
                },
                new MessageHandlerOptions(ExceptionReceivedHandler) { MaxConcurrentCalls = 10,AutoComplete = false });
        }
        private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs) {
            var ex = exceptionReceivedEventArgs.Exception;
            var context = exceptionReceivedEventArgs.ExceptionReceivedContext;
            _logger.LogError(ex,"ERROR handling message: {ExceptionMessage} - Context: {@ExceptionContext}",ex.Message,context);
            return Task.CompletedTask;
        }
        private async Task<bool> ProcessEvent(string eventName,string message) {
            var processed = false;
            if(_subsManager.HasSubscriptionsForEvent(eventName)) {
                using(var scope = _serviceScopeFactory.CreateScope()) {
                    var subscriptions = _subsManager.GetHandlersForEvent(eventName);
                    foreach(var subscription in subscriptions) {
                        if(subscription.IsDynamic) {
                            if(!(scope.ServiceProvider.GetService(subscription.HandlerType) is
                               IDynamicIntegrationEventHandler handler)) {
                                continue;
                            }
                            using dynamic eventData = JsonDocument.Parse(message);
                            await Task.Yield();
                            await handler.Handle(eventData);
                        } else {
                            var handler = scope.ServiceProvider.GetService(subscription.HandlerType);
                            if(handler == null) {
                                continue;
                            }
                            var eventType = _subsManager.GetEventTypeByName(eventName);
                            var integrationEvent = JsonSerializer.Deserialize(message,eventType,
                                new JsonSerializerOptions {
                                    PropertyNameCaseInsensitive = true
                                });
                            var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType);
                            await Task.Yield();
                            await (Task)concreteType.GetMethod("Handle").Invoke(handler,new[]
                            {
                                integrationEvent
                            });
                        }
                    }
                }
                processed = true;
            }
            return processed;
        }
        private void RemoveDefaultRule() {
            try {
                _serviceBusPersisterConnection
                    .SubscriptionClient
                    .RemoveRuleAsync(RuleDescription.DefaultRuleName)
                    .GetAwaiter()
                    .GetResult();
            } catch(MessagingEntityNotFoundException) {
                _logger.LogWarning("The messaging entity {DefaultRuleName} Could not be found.",RuleDescription.DefaultRuleName);
            }
        }
    }
}